package felix.things.router.mqtt.server.broker.service;

import com.alibaba.fastjson.JSONObject;
import felix.things.router.mqtt.server.broker.config.BrokerProperties;
import felix.things.router.mqtt.server.broker.internal.InternalMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author Felix
 * @date 2020/6/14
 */
@Component
@Slf4j
public class KafkaService {
    @Resource
    private KafkaTemplate<Object, Object> template;
    @Autowired
    private BrokerProperties brokerProperties;

    @Async
    public void send(InternalMessage internalMessage) {
        try {
            //消息体转换为Hex字符串进行转发
            template.send(brokerProperties.getProducerTopic(), internalMessage.getTopic(), JSONObject.toJSONString(internalMessage))
                    .addCallback(success -> {
                        log.info("The offset of the record we just sent is: " + success);
                    }, throwable -> {
                        log.error(throwable.getMessage(), throwable);
                    });
        } catch (Exception e) {
            log.error("kafka没有连接成功..");
        }
    }

}
